﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Text;
using System.Threading.Tasks;
using UtilZ.Dotnet.Ex.Base;
using UtilZ.Dotnet.Ex.Communication.Base;
using UtilZ.Dotnet.Ex.Communication.Net.Base;
using UtilZ.Dotnet.Ex.ZLog;

namespace UtilZ.Dotnet.Ex.Communication.Net.BMQ
{
    /// <summary>
    /// BMQ订阅器
    /// </summary>
    public class BMQSubscriber : IDisposable
    {
        private readonly BMQSubscriberConfig _config;
        /// <summary>
        /// 获取配置
        /// </summary>
        public BMQSubscriberConfig Config
        {
            get { return _config; }
        }

        private readonly ObjectEventWaitHandleManager _waitHandleManager;
        private readonly IUnpacker _unpacker;
        private readonly IClient _client;
        private readonly Dictionary<short, BMQSubscribeInfo> _subscribeInfoDic = new Dictionary<short, BMQSubscribeInfo>();
        private readonly object _lock = new object();
        private readonly ThreadEx _heartThread;


        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="config">配置</param>

        public BMQSubscriber(BMQSubscriberConfig config)
        {
            if (config == null)
            {
                throw new ArgumentNullException(nameof(config));
            }
            this._config = config;
            this._waitHandleManager = new ObjectEventWaitHandleManager();

            switch (config.Protocol)
            {
                case NetProtocols.Tcp:
                    this._unpacker = new BMQTcpTLVUnpacker();
                    var tcpIntiPara = new TcpClientInitPara(this._unpacker, config.PublisherEndPoint);
                    var tcpClient = new NetTcpClient(tcpIntiPara);
                    tcpClient.ConnectChanged += Client_ConnectChanged;
                    tcpClient.ConnectAsync();
                    this._client = tcpClient;
                    break;
                case NetProtocols.Udp:
                    this._unpacker = new BMQUDPUnpacker();
                    var udpIntiPara = new UdpClientInitPara(this._unpacker, config.BindEndPoint);
                    var udpClient = new NetUdpClient(udpIntiPara);
                    udpClient.Start();
                    this._client = udpClient;
                    break;
                default:
                    throw new NotSupportedException($"不支持协议{config.Protocol.ToString()}");
            }
            this._unpacker.PackOutput += Process_PackOutput;

            this._heartThread = new ThreadEx(this.HeartThreadMethod, "BMQSubscriber心跳线程");
            this._heartThread.Start();
        }


        private void HeartThreadMethod(ThreadExPara threadPara)
        {
            try
            {
                BMQHeartResMessage heartResMessage;
                while (!threadPara.Token.IsCancellationRequested)
                {
                    try
                    {
                        if (threadPara.WaitOne(this._config.HeartIntervalMilliseconds))
                        {
                            continue;
                        }

                        var heartReqMessage = new BMQHeartReqMessage(TimeEx.GetTimestamp());
                        heartResMessage = (BMQHeartResMessage)this.SendSessionMesage(heartReqMessage);
                    }
                    catch (TimeoutException tex)
                    {
                        ZLoger.Warn(tex);
                        //心跳超时，重新订阅
                        this.ReAllSubscribe();
                    }
                    catch (Exception exi)
                    {
                        ZLoger.Error(exi, "BQM订阅器心跳线程异常");
                    }
                }
            }
            catch (Exception ex)
            {
                ZLoger.Error(ex, "BQM订阅器心跳线程方法异常");
            }
        }

        private void ReAllSubscribe()
        {
            try
            {
                lock (this._lock)
                {
                    foreach (var subInfo in this._subscribeInfoDic.Values)
                    {
                        var subReqMessage = new BMQSubReqMessage(subInfo.Topic);
                        this.SendSessionMesage(subReqMessage);
                    }
                }
            }
            catch (Exception ex)
            {
                ZLoger.Error(ex, "BMQ全部重新订阅失败");
            }
        }

        private void Client_ConnectChanged(object sender, NetConnectedChangedArgs e)
        {
            if (e.Status == ConnectionStatus.Opened)
            {
                ZLoger.Info("BMQ-tcp连接已建立...");
                //重新建立连接，重新订阅
                this.ReAllSubscribe();
            }
            else
            {
                ZLoger.Warn("BMQ-tcp连接已断开...");
            }
        }

        private void Process_PackOutput(object sender, DataOutputArgs e)
        {
            try
            {
                byte[] buffer = e.Data.Bytes;
                byte messageType = buffer.First();
                switch (messageType)
                {
                    case BMQMessageTypes.SubRes:
                        ZLoger.Debug("收到订阅请求响应...");
                        this.ProcessSessionResMesage<BMQSubResMessage>(e);
                        ZLoger.Debug("处理收到订阅请求响应完成...");
                        break;
                    case BMQMessageTypes.UnsubRes:
                        ZLoger.Debug("收到取消订阅请求响应...");
                        this.ProcessSessionResMesage<BMQUnsubResMessage>(e);
                        ZLoger.Debug("处理收到取消订阅请求响应完成...");
                        break;
                    case BMQMessageTypes.HeartReq:
                        ZLoger.Debug("收到心跳请求...");
                        //收到心跳消息解析并回复
                        var heartReqMessage = new BMQHeartReqMessage();
                        heartReqMessage.FromBytes(buffer);
                        var heartResMessage = new BMQHeartResMessage(heartReqMessage.ContextId, heartReqMessage.Timestamp, TimeEx.GetTimestamp());
                        this.SendMesage(heartResMessage);
                        ZLoger.Debug("处理完成心跳请求...");
                        break;
                    case BMQMessageTypes.HeartRes:
                        ZLoger.Debug("收到心跳响应请求...");
                        this.ProcessSessionResMesage<BMQHeartResMessage>(e);
                        ZLoger.Debug("处理完成心跳响应请求...");
                        break;
                    case BMQMessageTypes.TextMessage:
                        ZLoger.Debug("收到文本消息...");
                        this.ProcessDataMesage<BMQTextMessage>(e);
                        break;
                    case BMQMessageTypes.BlobMessage:
                        ZLoger.Debug("收到二进制消息...");
                        this.ProcessDataMesage<BMQBlobMessage>(e);
                        break;
                    default:
                        ZLoger.Warn($"未处理的BMQ消息类型\"{messageType}\"");
                        break;
                }
            }
            catch (Exception ex)
            {
                ZLoger.Error(ex, "处理BMQ消息异常");
            }
        }

        private void ProcessDataMesage<T>(DataOutputArgs e) where T : IBMQDataMessage, new()
        {
            IBMQDataMessage message = new T();
            message.FromBytes(e.Data.Bytes);

            lock (this._lock)
            {
                BMQSubscribeInfo subscribInfo;
                if (this._subscribeInfoDic.TryGetValue(message.Topic, out subscribInfo))
                {
                    subscribInfo.OnRaiseMessageOutput(message);
                }
                else
                {
                    ZLoger.Warn($"主题\"{message.Topic}\"未订阅");
                }
            }
        }

        private void ProcessSessionResMesage<T>(DataOutputArgs e) where T : IBMQSessionMessage, new()
        {
            IBMQSessionMessage message = new T();
            message.FromBytes(e.Data.Bytes);
            EventWaitHandleInfo waitHandleInfo = this._waitHandleManager.GetEventWaitHandleInfo(message.ContextId);
            if (waitHandleInfo != null)
            {
                waitHandleInfo.Set(message);
            }
            else
            {
                ZLoger.Warn($"收到超时的消息:{message.ToString()},传话标识:{message.ContextId}");
            }
        }


        private void SendMesage(IBMQMessage message)
        {
            byte[] buffer;
            switch (this._config.Protocol)
            {
                case NetProtocols.Tcp:
                    buffer = message.ToTLVBytes();
                    ((NetTcpClient)this._client).Send(buffer);
                    break;
                case NetProtocols.Udp:
                    buffer = message.ToBytes();
                    ((NetUdpClient)this._client).SendTo(buffer, this._config.PublisherEndPoint);
                    break;
                default:
                    throw new NotSupportedException($"不支持协议{this._config.Protocol.ToString()}");
            }
        }

        private IBMQSessionMessage SendSessionMesage(IBMQSessionMessage message)
        {
            int timeoutMilliseconds = this._config.TimeoutMilliseconds;
            //100是为确定超时后，一定不会处理收到的消息
            using (EventWaitHandleInfo waitHandleInfo = this._waitHandleManager.Create(message.ContextId, timeoutMilliseconds + 100))
            {
                this.SendMesage(message);
                if (waitHandleInfo.WaitOne(timeoutMilliseconds))
                {
                    this._waitHandleManager.RemoveEventWaitHandle(message.ContextId);
                    return (IBMQSessionMessage)waitHandleInfo.Tag;
                }
                else
                {
                    throw new TimeoutException($"BMQ传话消息类型\"{message.ToString()}\"传话\"{message.ContextId}\"响应超时...");
                }
            }
        }


        /// <summary>
        /// 订阅
        /// </summary>
        /// <param name="subscribInfo">订阅对象</param>
        public void Subscribe(BMQSubscribeInfo subscribInfo)
        {
            lock (this._lock)
            {
                if (this._subscribeInfoDic.ContainsKey(subscribInfo.Topic))
                {
                    throw new ArgumentException($"主题\"{subscribInfo.Topic}\"已订阅");
                }

                var subReqMessage = new BMQSubReqMessage(subscribInfo.Topic);
                this.SendSessionMesage(subReqMessage);
                this._subscribeInfoDic.Add(subscribInfo.Topic, subscribInfo);
            }
        }


        /// <summary>
        /// 取消订阅,返回取消订阅信息,未订阅返回null
        /// </summary>
        /// <param name="topic">订阅主题</param>
        /// <returns>返回取消订阅信息,未订阅返回null</returns>
        public BMQSubscribeInfo UnSubscribe(short topic)
        {
            lock (this._lock)
            {
                BMQSubscribeInfo subscribInfo;
                if (!this._subscribeInfoDic.TryGetValue(topic, out subscribInfo))
                {
                    ZLoger.Warn($"主题\"{topic}\"未订阅");
                    return null;
                }

                var subReqMessage = new BMQUnsubReqMessage(topic);
                this.SendSessionMesage(subReqMessage);
                this._subscribeInfoDic.Remove(topic);
                return subscribInfo;
            }
        }


        /// <summary>
        /// 清空订阅
        /// </summary>
        public BMQSubscribeInfo[] Clear()
        {
            lock (this._lock)
            {
                foreach (var topic in this._subscribeInfoDic.Keys)
                {
                    var subReqMessage = new BMQUnsubReqMessage(topic);
                    this.SendSessionMesage(subReqMessage);
                }
                var subscribInfos = this._subscribeInfoDic.Values.ToArray();
                this._subscribeInfoDic.Clear();
                return subscribInfos;
            }
        }




        /// <summary>
        /// 对象释放标识[true:已释放;fasle:未已释放]
        /// </summary>
        protected bool _disposed = false;
        /// <summary>
        /// Dispose
        /// </summary>
        public void Dispose()
        {
            if (this._disposed)
            {
                return;
            }
            this._disposed = true;

            try
            {
                this.Clear();
                this._client.Dispose();
                this._unpacker.Dispose();
                this._waitHandleManager.Dispose();
            }
            catch (Exception ex)
            {
                ZLoger.Error(ex);
            }
        }

    }
}
